package server;
import java.io.*;
import java.net.*;
import java.util.Iterator;
import java.util.Vector;
import java.util.concurrent.Semaphore;

/**
 * @author dinganw
 *
 */
class serverThread implements Runnable {   
	//socket parameters
	private Socket server;
//	final static int BUF_LEN= 4096; 
	final static int TIMEOUT = 10000;//failstop failure of client: timeout =30s
	final static int EXP_FIELDS = 3;//the command should have at least 3 fields
	
	final static Semaphore mutex = new Semaphore(1,true);//to build semaphore for commit command
	
	//logs can be stored in two ways, a log is in the txntable(in memory)
	//Another log will be stored in log files
	static Vector<TXN> txntable=new Vector<TXN>();
	static int i=-1;//to counting next generating transaction number
	
	//socket variables
	private BufferedReader socketIn=null;
	private DataOutputStream socketOut = null;
	
	//read streamfiles variables
	File f=null;
	private DataInputStream fileIn = null;
	
	//to store and process request messageHead and data
	private static String messageHead = "";
	private static String data="";		
	private static String filePath;
		
	String[] result;
	private String method="", txn_id = "-1", seq_num = "0";
	private String content_length = "0";
		
	//the response message variables
	private String reasonContent_length = "";
	private String reason="";	
	private String responseMessage;
	
    serverThread(Socket server)
    {
      this.server=server;
    }
     
    public String getFilePath(){
    	return filePath;
    } 
    
    public static void setFilePath(String _filePath){
    	filePath=_filePath;
    }
    
    public void run ()
    {	
		try{
		System.out.println("Server is receiving.....");
	    socketIn = new BufferedReader (new InputStreamReader(new BufferedInputStream(server.getInputStream())));
//	    server.setSoTimeout(TIMEOUT);//set timeout for this transaction
	    messageHead=socketIn.readLine()+socketIn.readLine();				
		data=socketIn.readLine();	
		//parseClientCommand and do corresponding action		
		if(messageHead.length()==0){//wrong format
			reason="Received a blank message ~_~";
			System.out.println(reason);
			responseMessage=errorMessageReady(204);
			sendToClient(responseMessage);	
		}else{
			processClientCommand();	
			}//end if else
	    //end while				
	    }catch(SocketException se){
			if(se.getMessage().equals("Connection reset"))
				return;
			else{
				System.out.println("Could not read from or write to the socket: " + se.getMessage());
				return;
			}
		}catch(SocketTimeoutException ste){
			//time out, abort current transaction and keep reading message
			responseMessage=errorMessageReady(207);
			sendToClient(responseMessage);
			return ;
		}catch( IOException ioe){
			System.out.println("Could not read buffer: " + ioe.getMessage());
			return;
			}
    }
    
    private void processClientCommand(){   	
		result = messageHead.split("\\s+");		
		// to check the message format
		if(result.length < EXP_FIELDS)
		{	
//			reason="Wrong format, message head should have at least 3 fields!";
//			System.out.println(reason);
//			responseMessage=errorMessageReady(204);
//			sendToClient(responseMessage);
			return ;
		}	
		
		System.out.println("===================================");
		System.out.println("====>RECEIVED from Client:");
		System.out.println(messageHead);
		System.out.println(data);
		System.out.println("======================> end message");
		
			method = result[0];
			txn_id = result[1];
			seq_num = result[2];		
			content_length = result[3];	
		
		if(isMessageHeadFormatCorrect()==false){
			responseMessage=errorMessageReady(204);
			sendToClient(responseMessage);
			return ;
		}
	
		try {
			if(method.equals("READ"))		read();
			if(method.equals("NEW_TXN")) 	newTXN();
			if(method.equals("WRITE"))  	write();
			if(method.equals("COMMIT"))  	commit();
			if(method.equals("ABORT"))  	abort();
		} catch (IOException e) {
			responseMessage=errorMessageReady(205);
			sendToClient(responseMessage);
			return;
		}
		
		return ;   	
    }//end processClientCommand
  
    private void read() throws IOException{	 	
    	String fileName=data;
    	String fileRead;
    	
		// to check whether the file path is correct
		if(isFilePathCorrect()!=true){
			responseMessage=errorMessageReady(204);
			sendToClient(responseMessage);
			return ;	
		}// end if
		
		try{
		f= new File(filePath,fileName);
		byte[] buffer=new byte[(int) f.length()];
		FileInputStream fr= new FileInputStream(f);
		fileIn = new DataInputStream(fr);
		fileIn.read(buffer);
		fileRead=new String(buffer);	
		responseMessage=fileRead;
		sendToClient(responseMessage);
		
		}catch(FileNotFoundException err){
		//file not found
			responseMessage=errorMessageReady(206);
			sendToClient(responseMessage);
		}	
		
		return ;	
	 }
   
 	private void newTXN()throws IOException{
 		String fileName=data;
 		method="ACK";
 		txn_id=new Integer(++i).toString(); //generate this id for new transaction
 		TXN txn = new TXN(i,messageHead,filePath,fileName);	//record transaction
		txntable.add(txn);// keep transaction in the vector	
		writeLastTXNNum();
		txn.writeLog();
 		sendResponseMessage();
	    return ;
    } 
   
    private void write(){  
    	int txnID =(new Integer(txn_id)).intValue();
    	TXN writeTXN = findTXN(txnID);
    	
    	//to check errors of 201 and 202
    	if(writeTXN==null){
    		sendToClient(errorMessageReady(201));
    		return ;
    	}
    	if(writeTXN.dataMapToDisk.get(seq_num) != null){
    	// there is a sequence in the map
    		sendToClient(errorMessageReady(202));
    		return ;
    	}
    	
    	//to add this message into dataMapToDisk
    	writeTXN.addData(seq_num,data);  		 	 	
	    if(writeTXN.totalNumber%3==0){
	    	method="ACK";
	    	writeLastTXNNum();
	    	writeTXN.writeLog();
	    	sendResponseMessage();
    	}//end if
    	
    	return ;
    }
    
    private void commit() throws IOException{
    	int txnID = (new Integer(txn_id)).intValue();
    	TXN commitTXN = findTXN(txnID);
    	
    	//did not finding it,false transactioin ID, errors of 201, 
    	if(commitTXN==null){
    		sendToClient(errorMessageReady(201));
    		return ;
    	}
    	
    	int totalNum=(new Integer(seq_num)).intValue();	
    	
    	//partial commit - report error, e.g. client did 11 writes, but sends COMMIT message with 10 writes
    	if(totalNum<commitTXN.totalNumber){
    		seq_num=new Integer(commitTXN.totalNumber).toString();
    		sendToClient(errorMessageReady(202));
    	}else if(commitTXN.totalNumber==totalNum)
    	{
    	//successfully commit, sequentially commit, to acquire semaphore before commit
    		try {
			  mutex.acquire();
			  try {
				  commitTXN.commit();
			  } finally {
			    mutex.release();
			  }
			} catch(InterruptedException ie) {
				ie.printStackTrace();
    		}			
			method="ACK";
			//write log
			writeLastTXNNum();
			commitTXN.writeLog();
			sendResponseMessage();
			//after sending response, remove the committing transaction of current logs in txntable
			//and the logs in files
			txntable.remove(commitTXN);
			commitTXN.delelteLog();
    	}else{
    	//need to ask client for resending
    		method="ASK_RESEND";
    		String resendSeq_num=commitTXN.findResendSeq(totalNum);
    		seq_num=resendSeq_num;
    		sendResponseMessage();
    	}
    	//response to client
    	
    	return ;
    }
    
    private void abort(){
    	int txnID = (new Integer(txn_id)).intValue();
    	TXN abortTXN = findTXN(txnID);
    	
    	//to check errors of 201, false transactioin ID
    	if(abortTXN==null){
    		sendToClient(errorMessageReady(201));
    	}else{	
    		method="ACK";
    		txntable.remove(abortTXN);
    		abortTXN.delelteLog();
    		sendResponseMessage();
    	}
    	return ;
    }
    
    //write total transaction number into log so that the server can restart.
    private void writeLastTXNNum(){
    	try{
			File toFile=new File(filePath,".TXNNum.log");
		if(toFile.canRead()!=true){
			//cannot open the file, so create a new file and set the createdAFile tag
			toFile.createNewFile();	
		}		
		FileOutputStream fout= new FileOutputStream(toFile);		
		ObjectOutputStream oos=new ObjectOutputStream(fout);
		oos.writeObject(new Integer(i).toString());
		oos.close();	
		}catch(Exception ex){
		   ex.printStackTrace();
		   System.out.println("Cannot write log.!!!");
		} 	
    }
    
    private void deleteLastTXNNum(){
		File toFile=new File(filePath,".TXNNum.log");
		toFile.delete();
	 	
    }
    //write total transaction number into log so that the server can restart.
    public int readLastTXNNum(){
    	int TXNNum=-1;
    	try{
		File toFile=new File(filePath,".TXNNum.log");
		if(toFile.canRead()!=true){
			//cannot open the file, so create a new file and set the createdAFile tag
			return TXNNum;
		}		
		FileInputStream fin= new FileInputStream(toFile);		
		ObjectInputStream input=new ObjectInputStream(fin);
		String temp=(String)input.readObject();
		TXNNum=new Integer(temp).intValue();
		fin.close();	
		}catch(Exception ex){
		   ex.printStackTrace();
		   System.out.println("Cannot read log.!!!");
		}
		i=TXNNum;
		return TXNNum;
    }
    
    public void readtxntableTorestart(){
    	int theLastTXNNum=readLastTXNNum();
    	int k=0;
    	TXN temp;
    		//to add all the log object to the txntable
			while(k<=theLastTXNNum){
				try{
				File toFile=new File(filePath,"."+(new Integer(k).toString())+".log");
				++k;
				if(toFile.canRead()!=true){				
					continue ;
				}		
				FileInputStream fin= new FileInputStream(toFile);		
				ObjectInputStream input=new ObjectInputStream(fin);
				temp=(TXN)input.readObject();
				txntable.add(temp);
//				System.out.println("The txntable["+k+"]");
//				System.out.println(temp);		
				fin.close();	
				}catch(Exception ex){
					ex.printStackTrace();
					System.out.println("Cannot read log.!!!");
				}	
			}//end while
    }
    
    public void exitToDeleteLog(){
    	for(int j=0;j<i;++j){
    	//find transaction log to delete it
    	TXN txn = new TXN(j,messageHead,filePath,"");
    	txn.delelteLog();
    	}		
    }
    
    private void timeoutAbort(){
    	int txnID = (new Integer(i)).intValue();
    	TXN abortTXN = findTXN(txnID);
    	//to check errors of 201, false transactioin ID
    	if(abortTXN!=null){
    		txntable.remove(abortTXN);
    	}
    	return ;
    }
   
    /**
     * @param txnID
     * @return -find it :the transaction object which containing this ID
     * 		   -no find it :
     */
    private TXN findTXN(int txnID){
    	Iterator<TXN>  itr=txntable.iterator();
    	while(itr.hasNext()){
    		TXN temp=itr.next();
    		if(temp.TXNID==txnID){
    			return temp;
    		}	
    	}
    	return null;
    }
    
    private void sendResponseMessage(){
    	
    	String str=method + " " + txn_id + " " + seq_num + " \r\n\r\n\r\n";
    	sendToClient(str);
    	return ;
    }
    
    private void sendToClient(String response){	
		// to send response to client
    	byte[] toSend =(response).getBytes();	; //send bytes to client
    	try  
		{
    	    socketOut = new DataOutputStream(server.getOutputStream());
			System.out.println("=======================");
			System.out.println("MESSAGE RESPONSE TO Client:");
			System.out.println("====> begin message");
			System.out.println(response);
			socketOut.write(toSend);
			server.shutdownOutput();
			System.out.println("====> end message");					
		}
		catch( IOException ioe)
		{
			System.out.println("Could not send message to the server: " + ioe.getMessage());
			return;
		}
		
		return ; 
    }
   
    //prepare for response message regarding different errors
    private String errorMessageReady(int errorCode){
    	String str="";    	
    	if(errorCode==201){
    		reason= "Invalid transaction ID.";	
    	}else if(errorCode==202){
    		reason= "Invalid operation. ";   		
    	}else if(errorCode==204){
    		reason= "Wrong message format."; 		
    	}else if(errorCode==205){
    		reason= "File I/O error";  	
    	}else if(errorCode==206){
    		reason= "File not found";   		   	
    	}else if(errorCode==207){
    		reason= "Time Out,the server just keep the transaction in log, please resume ASAP!"; 
    	}else{
    		reason= "Unkown reasons";   
    	}
    	method="error";
		reasonContent_length=new Integer(reason.length()).toString();
		str= method + " " + txn_id + " " + seq_num + " " +new Integer(errorCode).toString()+" "+reasonContent_length
				+"\r\n\r\n"+reason;
    	return str ;
    }//end errorReport
  
    //to check whether the file path is correct.
    //it is wrong to try to pass a directory or create a hidden files
    private boolean isFilePathCorrect(){
    	if(data.contains("/")){
    		return false;
    	}else if(data.startsWith(".")){
    		return false;
    	}else if(data.length()==0){
    		return false;
    	}else{
    		return true;
    	}   		
    }
    
    private boolean isMessageHeadFormatCorrect(){
    	/* Check if the method is valid */
		if(!method.equals("WRITE") &&!method.equals("write")
				&&!method.equals("NEW_TXN") &&!method.equals("new_txn")
				&&!method.equals("COMMIT") &&!method.equals("commit")
				&&!method.equals("ABORT") &&!method.equals("abort")
				&&!method.equals("READ")&&!method.equals("read"))
		{	return false;
		}else if(new Integer(seq_num).intValue()<0)	{
			return false;
		}else if(new Integer(content_length).intValue()<0){
			return false;
		}else{
			return true;
		}
    }
     
}//end class


